package com.shujia.streaming

import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo02UpdateStateByKey {
  def main(args: Array[String]): Unit = {
    /**
     * 如果需要对之前的状态进行维护
     * 例如 统计单词数量时 单词的数量能够一直累加
     * 而不是每个批次重新计算
     * 则需要使用updateStateByKey有状态算子
     */
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo02UpdateStateByKey")
    conf.setMaster("local[2]")
    val sc: SparkContext = new SparkContext(conf)
    // 每5s一个批次
    val ssc: StreamingContext = new StreamingContext(sc, Durations.seconds(5))

    // 如果使用了有状态算子则需要设置checkPoint的目录
    ssc.checkpoint("spark/data/ssc/ck")

    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)

    /**
     *
     * @param seq 当前批次中属于某一个Key的所有的Value
     * @param opt 上一个批次中属于同一个Key的状态
     * @return
     */
    def myUpdateStateFunc(seq: Seq[Int], opt: Option[Int]): Option[Int] = {
      val sumValue: Int = seq.sum // 将当前批次中某一个Key的所有的Value加在一起
      val lastValue: Int = opt.getOrElse(0)
      Some(sumValue + lastValue)
    }

    lineDS
      .flatMap(_.split(","))
      .map(word => (word, 1))
      //      .updateStateByKey(myUpdateStateFunc)
      .updateStateByKey((seq: Seq[Int], opt: Option[Int]) => {
        Some(seq.sum + opt.getOrElse(0))
      })
      .print()


    ssc.start() // 启动流任务
    ssc.awaitTermination() // 等待停止
    ssc.stop() // 停止任务
  }

}
